package com.nightsoul.hadoop1.test.counter;

import java.io.IOException;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.nightsoul.hadoop1.test.OldJobBuilder;
import com.nightsoul.hadoop1.test.junit.NcdcRecordParser;
import com.nightsoul.hadoop1.test.mapreduce.MaxTemperatureReduce;

public class MaxTemperatureWithCounter extends Configured implements Tool {
	enum Temperature {
		MISSING, MALFORMED
	}
	
	static class MaxTemperatureMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
		private NcdcRecordParser parser = new NcdcRecordParser();
		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, IntWritable> output, Reporter reporter)
				throws IOException {
			
			parser.parse(value);
			
			if(parser.isValidTemperature()) {
				int airTemperature = parser.getAirTemperature();
				output.collect(new Text(parser.getYear()), new IntWritable(airTemperature));
			} else if(parser.isMalformedTemperature()) {
				System.err.println("Ignoring possibly corrupt input:" + value);
				reporter.incrCounter(Temperature.MALFORMED, 1);
			} else if(parser.isMissingTemperature()) {
				reporter.incrCounter(Temperature.MISSING, 1);
			}
			
			// dynamic counter
			reporter.incrCounter("TemperatureQuality", parser.getQuality(), 1);
		}
		
	}
	
	@Override
	public int run(String[] args) throws Exception {
		JobConf conf = OldJobBuilder.parseInputAndOutput(this, getConf(), args);
		if(conf==null) {
			return -1;
		}
		
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(IntWritable.class);
		conf.setMapperClass(MaxTemperatureMapper.class);
		conf.setCombinerClass(MaxTemperatureReduce.class);
		conf.setReducerClass(MaxTemperatureReduce.class);
		
		JobClient.runJob(conf);
		
		return 0;
	}
	
	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new MaxTemperatureWithCounter(), args);
		System.exit(exitCode);
	}
}
